package com.butterfly.hotkey.common.disruptor.container;

import cn.hutool.core.thread.NamedThreadFactory;
import com.butterfly.hotkey.core.constant.Constants;
import com.butterfly.hotkey.common.disruptor.DisruptorProperties;
import com.butterfly.hotkey.common.disruptor.event.QueueEvent;
import com.butterfly.hotkey.common.disruptor.factory.QueueEventFactory;
import com.butterfly.hotkey.common.disruptor.processor.IQueueEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

/**
 * @author butterfly
 * @version 1.0
 * <p>
 * disruptor的容器
 */
public class DisruptorProcessorContainer implements IEventProcessorContainer {

    private static final Logger log = LoggerFactory.getLogger(Constants.WORKER);

    private final String name;

    private final IQueueEventProcessor queueEventProcessor;

    private final DisruptorProperties disruptorProperties;

    /**
     * 是否运行中
     */
    private boolean running;

    /**
     * disruptor
     */
    private Disruptor<QueueEvent> disruptor;

    /**
     * ringBuffer
     */
    private RingBuffer<QueueEvent> ringBuffer;

    public DisruptorProcessorContainer(String name,
                                       DisruptorProperties disruptorProperties,
                                       IQueueEventProcessor queueEventProcessor) {
        this.name = name;
        this.disruptorProperties = disruptorProperties;
        this.queueEventProcessor = queueEventProcessor;
    }

    @Override
    public void destroy() {
        this.stop();
    }

    @Override
    public void afterPropertiesSet() {
        QueueEventFactory factory = new QueueEventFactory();
        NamedThreadFactory threadFactory = new NamedThreadFactory("hotkey-disruptor-executor", true);
        disruptor = new Disruptor<>(factory, disruptorProperties.getBufferSize(), threadFactory, ProducerType.SINGLE, new YieldingWaitStrategy());
        disruptor.handleEventsWith((event, sequence, endOfBatch) -> queueEventProcessor.onEvent(event));
        ringBuffer = disruptor.getRingBuffer();
    }

    @Override
    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("container already running. " + this.name);
        }

        try {
            disruptor.start();
        } catch (Exception e) {
            throw new IllegalStateException("Failed to start disruptorProcessorContainer", e);
        }
        this.setRunning(true);

        log.info("running container: {}", this.name);
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    @Override
    public void stop() {
        if (this.isRunning()) {
            if (Objects.nonNull(disruptor)) {
                log.info("stop disruptor container,{}", this.name);
                disruptor.shutdown();
            }
            setRunning(false);
        }
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    @Override
    public void receiveQueueEvent(QueueEvent queueEvent) {
        // 获取下一个可用位置的下标
        long sequence = ringBuffer.next();
        // 返回可用位置的元素
        QueueEvent event = ringBuffer.get(sequence);
        // 设置该位置元素的值
        event.setData(queueEvent.getData());
        event.setSource(queueEvent.getSource());

        ringBuffer.publish(sequence);
    }
}
